package main

import (
	"bufio"
	"fmt"
	pipeline2 "gitee.com/jnshao/go-start/projects/pipelinedemo/pipeline"
	"os"
	"strconv"
)

const sourceFilename = "../large.in"
const resultFilename = "../large.out"

// const sourceFilename = "../small.in"
// const resultFilename = "../small.out"

// 单机版而言，并发使用channel效率肯定是下降的
// 好处，当文件过大，一台机器排不了，多机排序
func main() {
	//p, files := createPipeline(sourceFilename, 800000000, 4) // 4等分根据主机cup核比较优（一个节点管道认为是一个主机）   (800000000/64)/4=3,125,000
	// createNetworkPipeline(sourceFilename, 512, 4)
	// time.Sleep(time.Hour)
	p, files := createNetworkPipeline(sourceFilename, 800000000, 4) //平均每个文件读取int64数：800000000/8/4
	defer func() {
		for _, file := range files {
			file.Close()
		}
	}()
	writeToFile(p, resultFilename) //该方法运行，通道才真正打开
	printFile(resultFilename)
}

// 创建并行处理管道
// fileSize 文件字节数
// chunkCount 节点数 读取文件分块数
func createPipeline(filename string, fileSize, chunkCount int) (<-chan int, []*os.File) {

	chunkSize := fileSize / chunkCount //每个节点读取文件字节数
	//outs := make([]<-chan int, chunkCount)
	outs := []<-chan int{}
	files := []*os.File{}

	pipeline2.Init() //开始计时

	for count := 0; count < chunkCount; count++ {
		file, err := os.Open(filename) //这里file没有close，需要返回*[]File,在外面close
		if err != nil {
			panic(err)
		}
		files = append(files, file)

		//Seek设置下一次读/写的位置。offset为相对偏移量，
		//whence决定相对位置：0为相对文件开头，1为相对当前位置，2为相对文件结尾
		file.Seek(int64(count*chunkSize), 0) //读文件字节范围
		source := pipeline2.ReadSource(bufio.NewReader(file), chunkSize)
		outs = append(outs, pipeline2.InMemSort(source))
	}
	//构建管道，goroutine还没有运行，不能确定InMemSort是否全部排序完成，不能在该方法close file
	return pipeline2.MergeN(outs...), files
}

// 创建并行处理管道
// fileSize 文件字节数
// chunkCount 节点数 读取文件分块数
func createNetworkPipeline(filename string, fileSize, chunkCount int) (<-chan int, []*os.File) {

	chunkSize := fileSize / chunkCount //每个节点读取文件字节数

	//outs := make([]<-chan int, chunkCount)
	outs := []<-chan int{}
	sortAddr := []string{}

	files := []*os.File{}

	pipeline2.Init() //开始计时

	//#region 节点服务端工作

	for count := 0; count < chunkCount; count++ {
		file, err := os.Open(filename) //这里file没有close，需要返回*[]File,在外面close
		if err != nil {
			panic(err)
		}
		files = append(files, file)

		//Seek设置下一次读/写的位置。offset为相对偏移量，
		//whence决定相对位置：0为相对文件开头，1为相对当前位置，2为相对文件结尾
		file.Seek(int64(count*chunkSize), 0) //读文件字节范围

		source := pipeline2.ReadSource(bufio.NewReader(file), chunkSize)

		// outs = append(outs, pipeline.InMemSort(source))
		//本机地址
		addr := ":" + strconv.Itoa(7000+count)                   //将数字转换成对应的字符串类型的数字
		pipeline2.NetworkSink(addr, pipeline2.InMemSort(source)) //开启节点服务监听，收到请求发送数据将写入到Network,异步不能等待阻塞

		sortAddr = append(sortAddr, addr)
	}

	//#endregion

	// return nil, files

	//#region Network工作

	for _, addr := range sortAddr {
		outs = append(outs, pipeline2.NetworkSource(addr))
	}
	//构建管道，goroutine还没有运行，不能确定InMemSort是否全部排序完成，不能在该方法close file
	return pipeline2.MergeN(outs...), files

	//#endregion
}

func writeToFile(in <-chan int, filename string) {
	file, err := os.Create(filename)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	w := bufio.NewWriter(file)
	defer w.Flush()

	pipeline2.WriteSink(w, in)
}

func printFile(filename string) {
	file, err := os.Open(filename)
	if err != nil {
		panic(err)
	}
	defer file.Close()
	count := 0
	all := pipeline2.ReadSource(bufio.NewReader(file), -1)
	for s := range all {
		fmt.Println(s)
		count++
		if count > 100 {
			break
		}
	}
}
